package com.atguigu.flink.chapter01_wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.api.common.typeinfo.Types.*;



public class Demo7_WEBUI
{
    public static void main(String[] args) throws Exception {

        //设置本地webui的监控端口
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        //env.setParallelism(1);

        //使用环境读取数据源，获取一个流
        // The parallelism of non parallel operator must be 1.
        // 有些算子socketTextStream，是非并行算子，这些算子的 parallelism只能是1
        DataStreamSource<String> source = env.socketTextStream("hadoop103", 8888);//.setParallelism(2);


        source
            .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            })
            //.returns(new TypeHint<Tuple2<String, Integer>>() {})
            // Flink提供了一个工具类 Types,保存了常见的 TypeInfomation的实现
            .returns(TUPLE(STRING,INT)).setParallelism(2)
            .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
            .sum(1).setParallelism(4)
            .print().setParallelism(8);

        //启动执行环境，计算才会开始，永不结束
        env.execute();

    }
}
